package cn.initcap.concurrency.concurrency.mq.kafka;

import cn.initcap.concurrency.concurrency.mq.Message;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import java.util.Date;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

/**
 * @author initcap
 * @date Created in 2018/6/20 AM8:34.
 */
@Component
@Slf4j
public class KafKaSender {

    private KafkaTemplate<String, String> kafkaTemplate;
    private Gson gson = new GsonBuilder().create();

    @Autowired
    public KafKaSender(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void send(String msg) {
        Message message = new Message();
        message.setId(System.currentTimeMillis());
        message.setMsg(msg);
        message.setSendTime(new Date());
        log.info("send message:{}", message);
        kafkaTemplate.send(TopicConstants.TEST, gson.toJson(message));
    }

}
